package com.df.gateway.utils;

import com.alibaba.nacos.shaded.com.google.common.collect.Maps;
import com.df.base.entity.MessageWrapper;
import com.df.base.other.IGateWayStreamInterface;
import com.df.base.other.R;
import com.df.gateway.entity.Client;
import com.df.gateway.magager.ClientManager;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.common.utils.PojoUtils;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.service.GenericService;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Slf4j
public class DubboUtils {



    private static ApplicationConfig application = new ApplicationConfig("generic-reference");

    private static Map<String, GenericService> serviceCache = Maps.newConcurrentMap();
    private static Map<String, IGateWayStreamInterface> streamServiceCache = Maps.newConcurrentMap();
    private static Map<String, StreamWraper> streamWrapperCache = Maps.newConcurrentMap();
    private static Map<String, Object> lockCache = Maps.newConcurrentMap();

    /**
     * 获取泛化服务接口(如有缓存, 则从缓存取)
     * @param reqModel
     * @return
     */
    public static ReferenceConfig<IGateWayStreamInterface> fetchGenericService(String group, String version) {

        // 配置调用信息
        ReferenceConfig<IGateWayStreamInterface> reference = new ReferenceConfig<IGateWayStreamInterface>();
        reference.setApplication(application);
//        reference.setScopeModel(S);
        reference.setInterface("com.df.base.other.IGateWayStreamInterface");
        reference.setGroup(group);
        reference.setVersion(version);
//        reference.setGeneric(true); // 声明为泛化接口
//        reference.setGeneric("true"); // 声明为泛化接口


        return reference;
    }
    public static GenericService getGenericService(String interfaceName,String group,String version) {
//        group = group==null?"":group;
        // 从缓存中获取服务
        String serviceCacheKey = interfaceName + "-" + group + "-"
                + version;
        GenericService service = serviceCache.get(serviceCacheKey);
        if (service != null) {
            log.info("fetched generic service from cache");
            return service;
        }
        //TODO 多线程获取，需要加锁
        // 访问的时候是每次都负载均衡还是连接后就 一直用这个连接？
        // 配置调用信息
        ReferenceConfig<GenericService> reference = new ReferenceConfig<GenericService>();
//        reference.setScopeModel(S);
        //TODO 注册中心
//        reference.setRegistry();
//        reference.setInterface(interfaceName+":::provider:"+group);
        reference.setInterface(interfaceName);
        reference.setGroup(group);
        reference.setVersion(version);
//        reference.setGeneric(true); // 声明为泛化接口
        reference.setGeneric("true"); // 声明为泛化接口
        //TODO 设置负载均衡的方式
//        reference.setLoadbalance();

        // 获取对应服务
        service = reference.get();

        // 缓存
        serviceCache.put(serviceCacheKey, service);

        return service;
    }
    /**
     * 根据接口类名及方法名通过反射获取参数类型
     *
     * @param interfaceName 接口名
     * @param methodName    方法名
     * @return
     * @throws ClassNotFoundException
     */
    public static String[] getMethodParamType(String interfaceName,
                                              String methodName) throws ClassNotFoundException {
        String[] paramTypeList = null;

        // 创建类
        Class<?> clazz = Class.forName(interfaceName);
        // 获取所有的公共的方法
        Method[] methods = clazz.getMethods();
        for (Method method : methods) {
            if (method.getName().equals(methodName)) {

                Class<?>[] paramClassList = method.getParameterTypes();
                paramTypeList = new String[paramClassList.length];

                int i = 0;
                for (Class<?> className : paramClassList) {
                    paramTypeList[i] = className.getTypeName();
                    i++;
                }
                break;
            }
        }

        return paramTypeList;
    }


    /**
     * Dubbo泛化调用
     * @param service 服务名，实际为IGateWayRpcInterface服务接口所属的gourp
     * @param version 服务接口的版本
     * @param params 参数，一般包含具体的服务和方法，参数，一般为具体服务节点里的服务接口名和方法，返回统一为R，比如参数里：
     *               {provider:User,method:Login,params:...}  说明：provider为服务里具体接口的名，method则为接口里的哪个方法，params为参数
     *               也可以将provider,method合并生产一个唯一id，在具体服里映射即可，注意前后端要一致
     * @return
     */
    public static R rpc(String service,String version,Object params){
        R r = new R();
        try {
            //TODO 模块从前端消息中解析，然后从模块映射中获取
            GenericService genericService = DubboUtils.getGenericService("com.df.base.other.IGateWayRpcInterface", service, version);
            // 设置隐式参数
            // TODO 设置隐式参数，然后服务提供者就可以在服务实现类方法里获取参数值，内部采用threadLocal实现
//            RpcContext.getContext().setAttachment("company", "alibaba");
            //TODO 方法可以映射具体的功能名字，从前端中获取（非http可以通过功能映射id的方式减少传输内容，防止被抓包）
//            Object o = genericService.$invoke("rpc", new String[]{"com.df.base.other.RpcReq"}, new Object[]{params});
            Object o = genericService.$invoke("rpc", new String[]{"java.lang.Object"}, new Object[]{params});
            r = PojoUtils.mapToPojo((Map<String, Object>) o, R.class);
            System.out.println(o);
        }catch (Exception e){
            e.printStackTrace();
            r.setCode(501);
            r.setMsg("服务调用出错！");

        }
        return r;
    }

    public static CompletableFuture<Object> rpcAync(String service,String version,Object params){
        CompletableFuture<Object> o = CompletableFuture.supplyAsync(()->{
            try {
                //TODO 模块从前端消息中解析，然后从模块映射中获取
                GenericService genericService = DubboUtils.getGenericService("com.df.base.other.IGateWayRpcInterface", service, version);
                //TODO 方法可以映射具体的功能名字，从前端中获取（非http可以通过功能映射id的方式减少传输内容，防止被抓包）
                return genericService.$invokeAsync("rpc", new String[]{"java.lang.Object"}, new Object[]{params}).get();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        });

        return o;
    }

    public static IGateWayStreamInterface getStreamServer( String group, String version) {
//        group = group==null?"":group;
        // 从缓存中获取服务
        String serviceCacheKey = "com.df.base.other.IGateWayStreamInterface" + "-" + group + "-"
                + version;
        IGateWayStreamInterface service = streamServiceCache.get(serviceCacheKey);
        if (service != null) {
            log.info("fetched stream service from cache");
            return service;
        }
        //TODO 多线程获取，需要加锁
        // 访问的时候是每次都负载均衡还是连接后就 一直用这个连接？
        // 配置调用信息
        ReferenceConfig<IGateWayStreamInterface> reference = new ReferenceConfig<IGateWayStreamInterface>();
//        reference.setScopeModel(S);
        //TODO 注册中心
//        reference.setRegistry();
//        reference.setInterface(interfaceName+":::provider:"+group);
        reference.setInterface("com.df.base.other.IGateWayStreamInterface");
        reference.setGroup(group);
        reference.setVersion(version);
        reference.setGeneric(true); // 声明为泛化接口
//        reference.setGeneric("true"); // 声明为泛化接口
        //TODO 设置负载均衡的方式


        // 获取对应服务
        service = reference.get();

        // 缓存
        streamServiceCache.put(serviceCacheKey, service);

        return service;
    }

    @Data
    private static class StreamWraper{
        String key;
        StreamObserver<Object> req;
        StreamObserver<Object> replay;
    }

    public static void message(String service,String version,Object data){
        //TODO 消息放入异步队列，发送和接收消息分开，线程池来发和单线程发的性能情况查看

        try {
            //TODO 模块从前端消息中解析，然后从模块映射中获取
            IGateWayStreamInterface genericService = DubboUtils.getStreamServer( service, version);
            // 设置隐式参数
            // TODO 设置隐式参数，然后服务提供者就可以在服务实现类方法里获取参数值，内部采用threadLocal实现
//            RpcContext.getContext().setAttachment("company", "alibaba");
            //TODO 方法可以映射具体的功能名字，从前端中获取（非http可以通过功能映射id的方式减少传输内容，防止被抓包）
//            Object o = genericService.$invoke("rpc", new String[]{"com.df.base.other.RpcReq"}, new Object[]{params});
            //TODO 对应一个服务器只需要调用一次就行么，中间不发消息连接会断开么
            String serviceCacheKey = "com.df.base.other.IGateWayStreamInterface" + "-" + service + "-"
                    + version;

            StreamWraper wraper = null;
            if(streamWrapperCache.containsKey(serviceCacheKey)){
                wraper =  streamWrapperCache.get(serviceCacheKey);
            }
            if(wraper==null){
                wraper = new StreamWraper();
                wraper.replay = getStreamObserver();
                wraper.req = genericService.message(wraper.replay);
                streamWrapperCache.put(serviceCacheKey,wraper);
            }
            wraper.req.onNext(data);


        }catch (Exception e){

        }
    }

    public static StreamObserver getStreamObserver(){
        return new StreamObserver<Object>() {
            @Override
            public void onNext(Object data) {
                //TODO 将收到的消息放入收的队列
//                String token = data.getToken();
//                Client client = ClientManager.Instance().getClientByToken(token);
//                if(client==null){
//                    return;
//                }
                System.out.println(data);
                //根据token获取client
//                ClientManager.Instance().handleMsg(client,data);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("发生了错误！");
                throwable.printStackTrace();
            }

            @Override
            public void onCompleted() {

            }
        };
    }
}

